package com.xbai.spark.core.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * aggregateByKey 算子
  * 作用：在 kv 对的 RDD 中，按 key 将 value 进行分组合并，
  * 合并时，将每个 value 和初始值作为 seq 函数的参数，进行计算，返回的结果作为一个新的 kv 对，
  * 然后再将结果按照 key 进行合并，最后将每个分组的 value 传递给 combine 函数进行计算
  * （先将前两个 value进行计算，将返回结果和下一个 value 传给 combine 函数，以此类推），将 key 与计算结果作为一个新的 kv 对输出。
  *
  * 参数：(zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U)
  *   zeroValue：给每一个分区中的每一个key一个初始值；
  *   seqOp：函数用于在每一个分区中用初始值逐步迭代value；
  *   combOp：函数用于合并每个分区中的结果。
  *
  * 案例：创建一个pairRDD，取出每个分区相同key对应值的最大值，然后相加
  *
  * @author xbai
  * @Date 2020/12/30
  */
object Spark18_AggregateByKey {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("aggregateByKey")
    val sc = new SparkContext(conf)

    val rdd: RDD[(String, Int)] = sc.parallelize(List(("a",3),("a",2),("c",4),("b",3),("c",6),("c",8)),2)
    val aggregateByKeyRDD: RDD[(String, Int)] = rdd.aggregateByKey(0)(math.max, _ + _)
    aggregateByKeyRDD.collect().foreach(println)
  }
}
